package com.eshipenet.shipnet.dp_decompressor

import java.lang

import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema
import org.apache.kafka.clients.producer.ProducerRecord
import org.json4s._
import org.json4s.native.Serialization.write

object EdgeChannelDataPacketUncompressedSerializer {
  implicit lazy val formats: Formats = DefaultFormats
}

class EdgeChannelDataPacketUncompressedSerializer extends KafkaSerializationSchema[EdgeChannelDataPacketUncompressed] {

  import EdgeChannelDataPacketUncompressedSerializer._

  override def serialize(element: EdgeChannelDataPacketUncompressed,
                         timestamp: lang.Long): ProducerRecord[Array[Byte], Array[Byte]] = {
    val topic = "EdgeChannelDataPacketUncompressed2"
    val key   = element.id.getBytes
    val obj = element.aux.values ++
      Map("id" -> JString(element.id), "data" -> JString(new String(element.data)))
    val value = write(obj).getBytes
    println(s"value = ${new String(value)}")
    new ProducerRecord[Array[Byte], Array[Byte]](topic, key, value)
  }

}
